package com.neu.mapreduce.tweet;

import java.io.IOException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * @author anshuman kumar
 */
public class TagFinder {
	public static final int MAX_RECORD = 10;

	/**
	 * @author anshuman kumar
	 * 
	 */
	public static class TagFindMapper extends Mapper<Object, Text, Text, Text> {
		public static final String TAB = "\t";
		public static final String TAG = "#";

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] line = value.toString().split(TAB);
			if (line.length > 9) {
				String timestamp = line[2];
				String tweetTxt = line[9];

				if ((tweetTxt.indexOf(TAG) != -1) && (timestamp.indexOf(":") != -1)
				        && (timestamp.split(" ").length == 6)) {
					String tag = tweetTxt.substring(tweetTxt.indexOf(TAG));
					if (tag.indexOf(" ") != -1) {
						tag = tag.substring(0, tag.indexOf(" "));
					}
					Text timestamps = new Text();
					timestamps.set(timestamp + "*" + timestamp);
					Text word = new Text();
					word.set(tag);
					// emit(tag g, {t1*t2})
					context.write(word, timestamps);
				}
			}
		}
	}

	/**
	 * @author anshuman kumar/
	 * 
	 */
	public static class TagFindCombiner extends Reducer<Text, Text, Text, Text> {
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
			Date min = null;
			Date max = null;
			try {
				for (Text val : values) {
					String start = val.toString().split("\\*")[0];
					String end = val.toString().split("\\*")[1];
					Date startDate = Util.createDate(start);
					Date endDate = Util.createDate(end);
					if (min == null) {
						min = startDate;
					} else if (Util.compareDates(min, startDate) > 0) {
						min = startDate;
					}

					if (max == null) {
						max = endDate;
					} else if (Util.compareDates(max, endDate) < 0) {
						max = endDate;
					}
				}
				Text result = new Text();
				result.set(min + "*" + max);
				context.write(key, result);

			} catch (Exception e) {
				e.printStackTrace();
			}

		}
	}

	/**
	 * @author anshuman kumar
	 * 
	 */
	public static class TagFindReducer extends Reducer<Text, Text, Text, Text> {
		private Map<String, String> top10Tags = null; // <tag, {timespan:starttime,endtime}>

		public void setup(org.apache.hadoop.mapreduce.Reducer<Text, Text, Text, Text>.Context context)
		        throws IOException, InterruptedException {
			top10Tags = new Hashtable<String, String>();
		}

		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
			Date min = null;
			Date max = null;
			try {
				for (Text val : values) {
					Date startDate = Util.createDate(val.toString().split("\\*")[0]);
					Date endDate = Util.createDate(val.toString().split("\\*")[1]);

					if (min == null) {
						min = startDate;
					} else if (Util.compareDates(min, startDate) > 0) {
						min = startDate;
					}
					if (max == null) {
						max = endDate;
					} else if (Util.compareDates(max, endDate) < 0) {
						max = endDate;
					}
				}
				if (top10Tags.size() < MAX_RECORD) {
					top10Tags.put(key.toString(), "" + Util.differenceDates(min, max) + ":" + min + ":" + max);
				} else {
					updateCache(key.toString(), min, max);
				}
			} catch (ParseException e1) {
				e1.printStackTrace();
			}

		}

		private void updateCache(String tag, Date min, Date max) {
			Set<String> tags = top10Tags.keySet();
			String removeKey = null;
			for (String tag1 : tags) {
				int tsMax = Integer.parseInt(((String) top10Tags.get(tag1)).split(":")[0]);
				if (Util.differenceDates(min, max) > tsMax) {
					removeKey = tag1;
					break;
				}
			}
			if (removeKey != null) {
				top10Tags.remove(removeKey);
				top10Tags.put(tag, "" + Util.differenceDates(min, max) + ":" + min + ":" + max);
			}

		}

		protected void cleanup(org.apache.hadoop.mapreduce.Reducer<Text, Text, Text, Text>.Context context)
		        throws IOException, InterruptedException {

			List<Map.Entry<String, String>> as = new ArrayList<Map.Entry<String, String>>(top10Tags.entrySet());
			Collections.sort(as, new Comparator<Map.Entry<String, String>>() {

				@Override
				public int compare(Map.Entry<String, String> e1, Map.Entry<String, String> e2) {
					int first = Integer.parseInt(e1.getValue().split(":")[0]);
					int second = Integer.parseInt(e2.getValue().split(":")[0]);
					return (second - first);
				}
			});

			for (Entry<String, String> entry : as) {
				Text tagKey = new Text();
				tagKey.set(entry.getKey());
				Text result = new Text();
				result.set(entry.getValue());
				context.write(tagKey, result);
			}

		};

	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: com.neu.mapreduce.tweet.TagFinder <in> <out>");
			System.exit(2);
		}
		Job job = new Job(conf, "Tag emdis");
		job.setJarByClass(TagFinder.class);
		job.setMapperClass(TagFindMapper.class);
		job.setCombinerClass(TagFindCombiner.class);
		job.setReducerClass(TagFindReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
